# import sys

# sys.path.append("..")  # 也可以这样

# from db_tools import MysqlTool
# from pandas import read_sql
# from tools.base_tools import get_current_and_previous_day_strings
# import requests
# from loguru import logger

# from pandas import DataFrame
# from pipei_data.api_requests import get_details_api, cps_books_api, search_api


# # # 通用详情的解析方法
# # def get_details_api(action_name, api_base_url, item_id):
# #     res_desc = requests.get(api_base_url, params={"data_id": item_id})
# #     logger.debug(res_desc.url)
# #     if action_name == "article":
# #         data_info_desc = res_desc.json()["info"]
# #     else:
# #         data_info_desc = res_desc.json()["list"][0]

# #     item = {
# #         "detail_id": data_info_desc["id"],
# #         "info_title": data_info_desc["title"],
# #         "cover": data_info_desc["cover"],
# #         "category_id": data_info_desc["category_id"],
# #         "name": data_info_desc.get("name"),
# #     }

# #     return item


# # # cps书籍的解析方法
# # def cps_boos_api(action_name, api_base_url, item_id):
# #     headers = {"token": "6a52237ad51213d4827d4ab33edaef6c"}
# #     res = requests.get(api_base_url, headers=headers, params={"id": item_id})
# #     # 需要对数据进行处理
# #     res_data = res.json()
# #     item = {
# #         "detail_id": res_data["data"]["id"],
# #         "info_title": res_data["data"]["title"],
# #         "category_id": res_data["data"]["new_cate_id"],
# #         "cover": res_data["data"]["image_url"],
# #         "name": res_data["data"].get("cate_name"),
# #         # 缺一个分类的标题
# #     }
# #     return item


# # def search_api(create_time, api_base_url, user_id):
# #     print(create_time, api_base_url, user_id)


# class Pipei(MysqlTool):
#     def __init__(self) -> None:
#         super().__init__()

#         self.action_dict = {
#             "article": {
#                 "func": "get_details_api",
#                 "base_url": "https://scapi.tayunapi.com/funcLog/article",
#                 "table": "course_pay_info",
#             },
#             "activity": {
#                 "func": "get_details_api",
#                 "base_url": "https://scapi.tayunapi.com/funcLog/activity",
#                 "table": "course_pay_info",
#             },
#             "chatgpt": {
#                 "func": "get_details_api",
#                 "base_url": "https://scapi.tayunapi.com/funcLog/chatgpt",
#                 "table": "course_pay_info",
#             },
#             "contract": {
#                 "func": "get_details_api",
#                 "base_url": "https://scapi.tayunapi.com/funcLog/contract",
#                 "table": "course_pay_info",
#             },
#             "course": {
#                 "func": "get_details_api",
#                 "base_url": "https://scapi.tayunapi.com/funcLog/course",
#                 "table": "course_pay_info",
#             },
#             "video": {
#                 "func": "get_details_api",
#                 "base_url": "https://scapi.tayunapi.com/funcLog/video",
#                 "table": "course_pay_info",
#             },
#             "cps_book": {
#                 "func": "cps_boos_api",
#                 "base_url": "http://tuicps.tylahs.com/v1/book/getBookInfo",
#                 "table": "course_pay_info",
#             },
#             "bookcps": {
#                 "func": "cps_boos_api",
#                 "base_url": "http://tuicps.tylahs.com/v1/book/getBookInfo",
#                 "table": "course_pay_info",
#             },
#             "banner": {
#                 "func": "get_details_api",
#                 "base_url": "https://scapi.tayunapi.com/funcLog/banner",
#                 "table": "course_pay_info",
#             },
#             "search": {
#                 "func": "search_api",
#                 "base_url": "https://scapi.tayunapi.com/funcLog/searchList",
#                 "table": "course_pay_info",
#             },
#             # 其他行为的处理逻辑...,慢慢处理
#         }

#     def read_in_type_data(self, intype_list=[]):
#         intype_list_str = ",".join([str(i) for i in intype_list])
#         sql = (
#             f"SELECT * FROM pay WHERE in_type IN ({intype_list_str}) AND is_matched = 0 "
#             f"AND ((`from` = 'IOS' AND version_code >= 156) "
#             f"OR (`from` = '安卓' AND version_code >= 31))"
#         )
#         return read_sql(sql, self.engine)

#     def get_action_info(self, order_info):
#         days_filter = get_current_and_previous_day_strings(order_info["create_time"])
#         url = "https://scapi.tayunapi.com/funcLog/logs"

#         params = {
#             "start": days_filter[0],
#             "end": days_filter[1],
#             "page": 1,
#             "rows": 10,  # 过滤的信息
#             "user_id": order_info["user_id"],
#             "sort": "desc",
#             "orderby": "id",
#             "create_time": order_info["create_time"],
#         }
#         response = requests.get(url, params=params)
#         print(response.url)
#         return response.json()

#     def get_order_action(self, action_info, order_info):
#         item = {}
#         for action in action_info["list"]:
#             # details 走一种,其他的可以单独走一种
#             if action["action"] in self.action_dict:
#                 item = action
#                 break

#         # 如果是找不到情况要怎么处理?
#         item["order_no"] = order_info["order_no"]
#         item["order_key"] = order_info["key"]  # 订单取到的key
#         # 在这里做判断
#         return item

#     def get_api_desc(self, order_action_info, order_info):
#         # 这种情况下是匹配不到具体的详情数据的,这里就pass掉
#         item_id = order_action_info["item_id"]
#         if item_id == 0:
#             return -2

#         # 从这里获取到action的具体的名称
#         action_name = order_action_info["action"]

#         api_base_url = self.action_dict[action_name]["base_url"]
#         # 用字符串的方式来当做函数来用
#         if action_name == "search":
#             res_desc = globals()[self.action_dict[action_name]["func"]](
#                 order_action_info["create_time"],
#                 api_base_url,
#                 order_action_info["user_id"],
#             )
#         else:
#             res_desc = globals()[self.action_dict[action_name]["func"]](
#                 action_name, api_base_url, item_id
#             )
#         # 要返回具体的action,后面要知道映射到那张表上..
#         return {**res_desc, **order_action_info}

#     # 清洗上面的数据
#     def format_info_data(self, info_data):
#         keys = [
#             "order_no",
#             "detail_id",
#             "info_title",
#             "cover",
#             "id",
#             "module",
#             "action",
#             "item_custom",
#             "create_time",
#             "category_id",
#             "name",
#             "order_key",
#         ]

#         selected_data = {k: info_data[k] for k in keys if k in info_data}

#         # 把key进行重命名一下
#         if "create_time" in selected_data:
#             selected_data["action_create_time"] = selected_data.pop("create_time")

#         if "id" in selected_data:
#             selected_data["action_id"] = selected_data.pop("id")
#         return selected_data

#     # 这里需要再判断一下,是不是要更新key的值
#     def update_to_mysql(self, selected_data, table="course_pay_info"):
#         df_res = DataFrame([selected_data])
#         self.upsert_df(df_res, table_name=table, primary_key="order_no")
#         # 把 is_matched = 1 更新,用sql语句
#         return self.execute_sql(
#             f"UPDATE pay SET is_matched = 1 WHERE order_no = '{selected_data['order_no']}'"
#         )

#     def start(self):
#         # 这个需要逐一测试
#         # intype_list = [
#         #     1032,
#         #     1033,
#         #     1040,
#         #     1034,
#         #     1035,
#         #     1041,
#         #     1028,
#         #     1014,
#         #     1009,
#         #     1010,
#         #     1036,
#         #     1037,
#         #     1038
#         #     1039,
#         # ]  # 先对视频,素材包进行匹配

#         # 先把搜索的全部匹配一下看看
#         intype_list = [1008]
#         from api_push.base_push import get_df_in_type

#         df_intype = get_df_in_type()
#         df = self.read_in_type_data(intype_list)
#         print(f"一共需要更新{len(df)}条数据")
#         for i in range(len(df[:1])):
#             order_info = df.iloc[i].to_dict()
#             logger.info(f"正在更新{order_info['order_no']}")
#             try:
#                 action_info = self.get_action_info(order_info)
#                 # 获取对应acton的具体信息
#                 order_action_info = self.get_order_action(action_info, order_info)

#                 # 搜索的判断逻辑
#                 if order_action_info["action"] == "search":
#                     info_data = self.get_api_desc(order_action_info, order_info)

#                 # 如果是获取不到文章详情的情况下:
#                 elif not order_action_info.get("item_id"):
#                     info_data = -2
#                 else:
#                     info_data = self.get_api_desc(order_action_info, order_info)
#                 # 如果 info_data 是一个数字,一般就是匹配不到的情况
#                 if isinstance(info_data, (int, float)):
#                     self.execute_sql(
#                         f"UPDATE pay SET is_matched = -2 WHERE order_no = '{order_info['order_no']}'"
#                     )
#                     logger.info(f"无需匹配 -> {order_info['order_no']}")
#                 else:
#                     # 如果是其他的保存逻辑,要从这里来判断

#                     # 这里是详情的保存逻辑
#                     info_data = self.format_info_data(info_data)
#                     # 如果取到的真实行为和订单的key不一样,就改成行为点击的key
#                     if info_data["order_key"] != info_data["action"]:
#                         # 处理一些特殊的情况,key值不对等的情况
#                         if info_data["action"] == "cps_book":
#                             new_key_name = "bookcps"
#                         else:
#                             new_key_name = info_data["action"]

#                         intype_dict = (
#                             df_intype[df_intype["key"] == new_key_name]
#                             .iloc[0]
#                             .to_dict()
#                         )
#                         # print(intype_dict)

#                         sql = f"""
#                             UPDATE pay
#                             SET
#                                 `key` = '{new_key_name}',
#                                 in_type = '{intype_dict['id']}',
#                                 title = '{intype_dict['title']}'
#                             WHERE
#                                 order_no = '{info_data['order_no']}'
#                             """
#                         logger.info(self.execute_sql(sql))

#                     # 把数据更新到数据表
#                     res = self.update_to_mysql(
#                         info_data, table=self.action_dict[info_data["action"]]["table"]
#                     )

#                     logger.info(f"{info_data['order_no']} -> {res}")
#             except Exception as err:
#                 logger.error(f"{order_info['order_no']},处理失败,{err}")
#                 # import traceback

#                 # traceback.print_exc()
#                 self.execute_sql(
#                     f"UPDATE pay SET is_matched = -1 WHERE order_no = '{order_info['order_no']}'"
#                 )

from pipei_data.action_handler import ActionHandler
from api_push.base_push import get_df_in_type
from loguru import logger
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
from threading import Lock

# 线程安全的计数器
progress_lock = Lock()
progress_stats = {"success": 0, "error": 0, "total": 0, "processed": 0}


def process_single_order(order_info, df_intype, handler_class=ActionHandler):
    """
    处理单个订单的匹配逻辑（线程安全版本）
    
    Args:
        order_info (dict): 订单信息
        df_intype: 入口类型数据
        handler_class: 处理器类
    
    Returns:
        tuple: (success, order_no, message)
    """
    # 每个线程创建自己的handler实例，避免线程冲突
    handler = handler_class()
    order_no = order_info['order_no']
    
    try:
        handler.process_order_info(order_info, df_intype)
        
        # 线程安全地更新进度
        with progress_lock:
            progress_stats["success"] += 1
            progress_stats["processed"] += 1
            if progress_stats["processed"] % 10 == 0:  # 每10条记录输出一次进度
                logger.info(f"📊 进度: {progress_stats['processed']}/{progress_stats['total']} "
                          f"(成功: {progress_stats['success']}, 失败: {progress_stats['error']})")
        
        return True, order_no, "处理成功"
        
    except Exception as err:
        # 处理失败，更新数据库状态
        try:
            handler.execute_sql_update(order_no, -1)
        except Exception as db_err:
            logger.error(f"数据库更新失败 {order_no}: {db_err}")
        
        # 线程安全地更新进度
        with progress_lock:
            progress_stats["error"] += 1
            progress_stats["processed"] += 1
            
        return False, order_no, str(err)


def start_concurrent(max_workers=6):
    """
    并发版本的匹配处理函数
    
    Args:
        max_workers (int): 最大并发线程数，建议4-8个
    """
    logger.info(f"🚀 开始并发匹配处理 (并发数: {max_workers})")
    
    # 初始化数据
    handler = ActionHandler()
    intype_list = [
        1008, 1032, 1033, 1040, 1034, 1035, 1041, 1028, 
        1014, 1009, 1010, 1036, 1037, 1038, 1039,
    ]
    df_intype = get_df_in_type()
    df = handler.read_in_type_data(intype_list)
    
    total_count = len(df)
    logger.info(f"📋 一共需要更新 {total_count} 条数据")
    
    if total_count == 0:
        logger.info("✅ 没有需要处理的数据")
        return
    
    # 重置进度统计
    with progress_lock:
        progress_stats.update({"success": 0, "error": 0, "total": total_count, "processed": 0})
    
    start_time = time.time()
    
    # 使用线程池执行并发处理
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有任务
        future_to_order = {}
        for i in range(total_count):
            order_info = df.iloc[i].to_dict()
            future = executor.submit(process_single_order, order_info, df_intype)
            future_to_order[future] = order_info['order_no']
        
        logger.info(f"📦 已提交 {total_count} 个并发任务")
        
        # 收集结果
        completed_count = 0
        for future in as_completed(future_to_order):
            order_no = future_to_order[future]
            completed_count += 1
            
            try:
                success, returned_order_no, message = future.result()
                if success:
                    logger.debug(f"✅ {returned_order_no}: {message}")
                else:
                    logger.error(f"❌ {returned_order_no}: {message}")
            except Exception as exc:
                logger.error(f"💥 {order_no} 线程执行异常: {exc}")
                with progress_lock:
                    progress_stats["error"] += 1
                    progress_stats["processed"] += 1
    
    # 输出最终统计
    end_time = time.time()
    elapsed_time = end_time - start_time
    
    with progress_lock:
        success_count = progress_stats["success"]
        error_count = progress_stats["error"]
        success_rate = (success_count / total_count * 100) if total_count > 0 else 0
    
    logger.info(f"🎉 === 并发匹配处理完成 ===")
    logger.info(f"📊 处理统计:")
    logger.info(f"   总数量: {total_count}")
    logger.info(f"   成功: {success_count} ({success_rate:.1f}%)")
    logger.info(f"   失败: {error_count}")
    logger.info(f"   耗时: {elapsed_time:.1f}秒")
    logger.info(f"   平均: {elapsed_time/total_count:.2f}秒/条")
    
    return success_count, error_count


def start():
    """
    原始串行版本（保持兼容性）
    """
    logger.info("⚠️  使用串行处理模式，如需提升性能请使用 start_concurrent()")
    
    handler = ActionHandler()
    intype_list = [
        1008, 1032, 1033, 1040, 1034, 1035, 1041, 1028,
        1014, 1009, 1010, 1036, 1037, 1038, 1039,
    ]
    df_intype = get_df_in_type()
    df = handler.read_in_type_data(intype_list)
    logger.info(f"一共需要更新{len(df)}条数据")

    for i in range(len(df[:])):
        order_info = df.iloc[i].to_dict()
        logger.info(f"正在更新 {order_info['order_no']}")
        try:
            handler.process_order_info(order_info, df_intype)
        except Exception as err:
            logger.error(f"{order_info['order_no']},处理失败,{err}")
            handler.execute_sql_update(order_info["order_no"], -1)


if __name__ == "__main__":
    start()
